package org.itsolutions.javaBackup.buffers;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.itsolutions.javaBackup.definitions.Count.TEN;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import org.itsolutions.javaBackup.logger.ELogLevel;
import org.itsolutions.javaBackup.logger.ILogger;
import org.itsolutions.javaBackup.logger.LoggerHelper;
import org.itsolutions.javaBackup.spi.common.io.IBuffer;
import org.itsolutions.javaBackup.spi.common.io.IBufferInterceptor;
import org.itsolutions.javaBackup.spi.common.io.IBufferListener;

/**
 * Buffer kind (FIFO) structure.
 * @param <E> type of items
 * @author Tomas Fecko
 */
public final class JBBuffer<E> extends LinkedBlockingDeque<E> implements IBuffer<E> {

	private static final long 			serialVersionUID = -4228557504945037979L;
	private static final ILogger 		LOGGER = LoggerHelper.getLogger(JBBuffer.class);
	private final SharedBufferProvider	mSharedBuffer;
	private final String				mUID;
	private final AtomicLong			mNumberOfDiscardedEvents = new AtomicLong(0);
	private final List<IBufferListener>	mListeners = Collections.synchronizedList(new ArrayList<IBufferListener>(0));
	private final ReentrantLock			mInterceptorsLock = new ReentrantLock();
	private final List<IBufferInterceptor<E>> mBufferInterceptors = Collections.synchronizedList(new ArrayList<IBufferInterceptor<E>>(0));

	/**
	 * 
	 * @param pSize size
	 * @param pUID uid
	 * @param pSharedBuffer shared buffer
	 */
	protected JBBuffer(SharedBufferProvider pSharedBuffer, String pUID, int pSize) {
		super(pSize);
		mSharedBuffer = pSharedBuffer;
		mUID = pUID;
	}

	@Override
	public void dispose() {
		LOGGER.log(ELogLevel.TRACE, "Disposing the buffer.");
		clear();
		mSharedBuffer.dispose(mUID);
	}

	@Override
	protected void finalize() throws Throwable {
		super.finalize();
		dispose();
	}

	@Override
	public void registerBufferListener(IBufferListener pBufferListener) {
		mListeners.add(pBufferListener);
	}

	@Override
	public boolean unregisterBufferListener(IBufferListener pBufferListener) {
		return mListeners.remove(pBufferListener);
	}

	@Override
	public void addInterceptor(IBufferInterceptor<E> pInterceptor) {
		mInterceptorsLock.lock();
		try {
			mBufferInterceptors.add(pInterceptor);
		} finally {
			mInterceptorsLock.unlock();
		}
	}

	@Override
	public void removeInterceptor(IBufferInterceptor<E> pInterceptor) {
		mInterceptorsLock.lock();
		try {
			mBufferInterceptors.remove(pInterceptor);
		} finally {
			mInterceptorsLock.unlock();
		}
	}

	@Override
	public void removeInterceptor(IBufferInterceptor<E> pInterceptor, E... pItems) {
		mInterceptorsLock.lock();
		try {
			mBufferInterceptors.remove(pInterceptor);
			if (pItems != null && pItems.length > 0) {
				for (E item : pItems) {
					if (item == null) {
						continue;
					}
					if (!super.offer(item)) {
						mNumberOfDiscardedEvents.incrementAndGet();
						LOGGER.log(ELogLevel.WARN, "Problem during the interceptor removal, buffer is full, discarding:", item);
					}
				}
			}
		} finally {
			mInterceptorsLock.unlock();
		}
	}

	@Override
	public long getNumberOfDiscardedEvents() {
		return mNumberOfDiscardedEvents.get();
	}

	//////////////////////////////////////////////////////////////////////////////////////////////////////
	///////////////////////////////////////// DEQUEUE ////////////////////////////////////////////////////
	//////////////////////////////////////////////////////////////////////////////////////////////////////

	///////////////////////////// ADD/INSERT operations //////////////////////////////////////////////////

	@Override
	public boolean offerOrDiscardIntercepted(E pItem) {
		if (processInterceptors(mBufferInterceptors, pItem)) {
			boolean offer = offer(pItem);
			if (!offer) {
				mNumberOfDiscardedEvents.incrementAndGet();
			}
			processListeners();			
			return offer;
		} else {
			return false;
		}
	}

	@Override
	public boolean offerOrDiscardIntercepted(E pItem, long pTimeoutInMillis) {
		if (processInterceptors(mBufferInterceptors, pItem)) {
			try {
				boolean offer = offer(pItem, pTimeoutInMillis, MILLISECONDS);
				if (!offer) {
					mNumberOfDiscardedEvents.incrementAndGet();
				}
				processListeners();				
				return offer;
			} catch (InterruptedException e) {
				LOGGER.log(ELogLevel.ERROR, e);
				return false;
			}
		} else {
			return false;
		}
	}

	@Override
	public void addFirst(E pE) {
		super.addFirst(pE);
		processListeners();
	}

	@Override
	public void addLast(E pE) {
		super.addLast(pE);
		processListeners();
	}

	@Override
	public boolean offerFirst(E pE) {
		boolean offerFirst = super.offerFirst(pE);
		processListeners();
		return offerFirst;
	}

	@Override
	public boolean offerLast(E pE) {
		boolean offerLast = super.offerLast(pE);
		processListeners();
		return offerLast;
	}

	@Override
	public void putFirst(E pE) throws InterruptedException {
		super.putFirst(pE);
		processListeners();
	}

	@Override
	public void putLast(E pE) throws InterruptedException {
		super.putLast(pE);
		processListeners();
	}

	@Override
	public boolean offerFirst(E pE, long pTimeout, TimeUnit pUnit) throws InterruptedException {
		boolean offerFirst = super.offerFirst(pE, pTimeout, pUnit);
		processListeners();
		return offerFirst;
	}

	@Override
	public boolean offerLast(E pE, long pTimeout, TimeUnit pUnit) throws InterruptedException {
		boolean offerLast = super.offerLast(pE, pTimeout, pUnit);
		processListeners();
		return offerLast;
	}

	@Override
	public boolean add(E pE) {
		boolean add = super.add(pE);
		processListeners();
		return add;
	}

	@Override
	public boolean offer(E pE) {
		boolean offer = super.offer(pE);
		processListeners();
		return offer;
	}

	@Override
	public void put(E pE) throws InterruptedException {
		super.put(pE);
		processListeners();
	}

	@Override
	public boolean offer(E pE, long pTimeout, TimeUnit pUnit) throws InterruptedException {
		boolean offer = super.offer(pE, pTimeout, pUnit);
		processListeners();
		return offer;
	}

	@Override
	public void push(E pE) {
		super.push(pE);
		processListeners();
	}

	///////////////////////////// REMOVE/POLL operations //////////////////////////////////////////////////
	@Override
	public E removeFirst() {
		E removeFirst = super.removeFirst();
		processListeners();
		return removeFirst;
	}

	@Override
	public E removeLast() {
		E removeLast = super.removeLast();
		processListeners();
		return removeLast;
	}

	@Override
	public E pollFirst() {
		E pollFirst = super.pollFirst();
		processListeners();
		return pollFirst;
	}

	@Override
	public E pollLast() {
		E pollLast = super.pollLast();
		processListeners();
		return pollLast;
	}

	@Override
	public E takeFirst() throws InterruptedException {
		E takeFirst = super.takeFirst();
		processListeners();
		return takeFirst;
	}

	@Override
	public E takeLast() throws InterruptedException {
		E takeLast = super.takeLast();
		processListeners();
		return takeLast;
	}

	@Override
	public E pollFirst(long pTimeout, TimeUnit pUnit) throws InterruptedException {
		E pollFirst = super.pollFirst(pTimeout, pUnit);
		processListeners();
		return pollFirst;
	}

	@Override
	public E pollLast(long pTimeout, TimeUnit pUnit) throws InterruptedException {
		E pollLast = super.pollLast(pTimeout, pUnit);
		processListeners();
		return pollLast;
	}

	@Override
	public boolean removeFirstOccurrence(Object pO) {
		boolean removeFirstOccurrence = super.removeFirstOccurrence(pO);
		processListeners();
		return removeFirstOccurrence;
	}

	@Override
	public boolean removeLastOccurrence(Object pO) {
		boolean removeLastOccurrence = super.removeLastOccurrence(pO);
		processListeners();
		return removeLastOccurrence;
	}

	@Override
	public E remove() {
		E remove = super.remove();
		processListeners();
		return remove;
	}

	@Override
	public E poll() {
		E poll = super.poll();
		processListeners();
		return poll;
	}

	@Override
	public E take() throws InterruptedException {
		E take = super.take();
		processListeners();
		return take;
	}

	@Override
	public E poll(long pTimeout, TimeUnit pUnit) throws InterruptedException {
		E poll = super.poll(pTimeout, pUnit);
		processListeners();
		return poll;
	}

	@Override
	public E pop() {
		E pop = super.pop();
		processListeners();
		return pop;
	}

	@Override
	public boolean remove(Object pO) {
		boolean remove = super.remove(pO);
		processListeners();
		return remove;
	}

	/**
	 * 
	 */
	private void processListeners() {
		if (mListeners.isEmpty()) {
			return;
		}
		int remainingCapacity = remainingCapacity();
		int size = size();
		for (IBufferListener listener : mListeners) {
			try {
				if (remainingCapacity == 0) {
					listener.bufferFull(size, 0);
				} else if (remainingCapacity == size) {
					listener.bufferHalfFull(size, remainingCapacity);
				} else if (((size + remainingCapacity) / TEN) == remainingCapacity) {
					listener.bufferAlmostFull(size, remainingCapacity);
				}
			} catch (Exception e) {
				LOGGER.log(ELogLevel.ERROR, e);
			}
		}
	}

	/**
	 * 
	 * @param pItem item
	 * @return true, if all interceptors returns true
	 */
	private boolean processInterceptors(List<IBufferInterceptor<E>> pBufferInterceptors, E pItem) {
		if (pBufferInterceptors.isEmpty()) {
			return true;
		}
		mInterceptorsLock.lock();
		try {
			boolean returnValue = true;
			for (IBufferInterceptor<E> interceptor : pBufferInterceptors) {
				try {
					if (!interceptor.intercept(pItem)) {
						returnValue = false;
					}
				} catch (Exception e) {
					LOGGER.log(ELogLevel.ERROR, e);
				}
			}
			return returnValue;
		} finally {
			mInterceptorsLock.unlock();
		}
	}
}






